<html>
 <head>
  <meta charset="UTF-8">
 </head>
 <body>
  <h1 data-lake-id="gTDL9" id="gTDL9"><span data-lake-id="ud007f672" id="ud007f672">典型回答</span></h1>
  <p data-lake-id="u9f30da10" id="u9f30da10"><br></p>
  <p data-lake-id="u46d87552" id="u46d87552"><span data-lake-id="uf2e3f26d" id="uf2e3f26d">当我们作为一个消费发送方，如何保证我们给RabbitMQ发送的消息一定能发送成功，如何确保他一定能收到这个消息呢？</span></p>
  <p data-lake-id="u1832112b" id="u1832112b"><span data-lake-id="u3c8064a4" id="u3c8064a4">​</span><br></p>
  <p data-lake-id="ue65c1e99" id="ue65c1e99"><span data-lake-id="ue6af9b16" id="ue6af9b16">我们知道，RabbitMQ的消息最终时存储在Queue上的，而在Queue之前还要经过Exchange，那么这个过程中就有两个地方可能导致消息丢失。第一个是Producer到Exchange的过程，第二个是Exchange到Queue的过程。</span></p>
  <p data-lake-id="ub66ada0a" id="ub66ada0a"><span data-lake-id="ua1cff638" id="ua1cff638">​</span><br></p>
  <p data-lake-id="uc4a2e6db" id="uc4a2e6db"><img src="https://cdn.nlark.com/yuque/0/2023/png/5378072/1693634392644-7843a67d-4fb1-4da9-9809-9b190a1bd035.png?x-oss-process=image%2Fwatermark%2Ctype_d3F5LW1pY3JvaGVp%2Csize_27%2Ctext_SmF2YSA4IEd1IFA%3D%2Ccolor_FFFFFF%2Cshadow_50%2Ct_80%2Cg_se%2Cx_10%2Cy_10"></p>
  <p data-lake-id="ued823b89" id="ued823b89"><br></p>
  <p data-lake-id="ue0f45991" id="ue0f45991"><span data-lake-id="u275df06b" id="u275df06b">为了解决这个问题，有两种方案，一种是通过confirm机制，另外一种是事务机制，因为事务机制并不推荐，这里先介绍Confirm机制。</span></p>
  <p data-lake-id="ucfb92150" id="ucfb92150"><span data-lake-id="u14fa1b1a" id="u14fa1b1a">​</span><br></p>
  <p data-lake-id="u85e3289f" id="u85e3289f"><br></p>
  <p data-lake-id="u2f68f362" id="u2f68f362"><span data-lake-id="uce5fe542" id="uce5fe542">上面两个可能丢失的过程，都可以利用confirm机制，注册回调来监听是否成功。</span></p>
  <p data-lake-id="uee6f1723" id="uee6f1723"><br></p>
  <p data-lake-id="ufae826d4" id="ufae826d4"><strong><span data-lake-id="uc72e44aa" id="uc72e44aa">Publisher Confirm</span></strong><span data-lake-id="u1117af0f" id="u1117af0f">是一种机制，用于确保消息已经被Exchange成功接收和处理。一旦消息成功到达Exchange并被处理，RabbitMQ会向消息生产者发送确认信号（ACK）。如果由于某种原因（例如，Exchange不存在或路由键不匹配）消息无法被处理，RabbitMQ会向消息生产者发送否认信号（NACK）。</span></p>
  <p data-lake-id="udc554d56" id="udc554d56"><span data-lake-id="u28bca0ed" id="u28bca0ed">​</span><br></p>
  <pre lang="java"><code>
// 启用Publisher Confirms
channel.confirmSelect();

// 设置Publisher Confirms回调
channel.addConfirmListener(new ConfirmListener() {
    @Override
    public void handleAck(long deliveryTag, boolean multiple) throws IOException {
        System.out.println("Message confirmed with deliveryTag: " + deliveryTag);
        // 在这里处理消息确认
    }

    @Override
    public void handleNack(long deliveryTag, boolean multiple) throws IOException {
        System.out.println("Message not confirmed with deliveryTag: " + deliveryTag);
        // 在这里处理消息未确认
    }
});

</code></pre>
  <p data-lake-id="ufce0d4f7" id="ufce0d4f7"><br></p>
  <p data-lake-id="ued1fbbb7" id="ued1fbbb7"><strong><span data-lake-id="u20a217e5" id="u20a217e5">Publisher Returns</span></strong><span data-lake-id="uf7284bd9" id="uf7284bd9">机制与Publisher Confirms类似，但用于处理在消息无法路由到任何队列时的情况。当RabbitMQ在无法路由消息时将消息返回给消息生产者，但是如果能正确路由，则不会返回消息。</span></p>
  <p data-lake-id="ucee6a7a6" id="ucee6a7a6"><span data-lake-id="u82b4442e" id="u82b4442e">​</span><br></p>
  <pre lang="java"><code>
// 启用Publisher Returns
channel.addReturnListener(new ReturnListener() {
    @Override
    public void handleReturn(int replyCode, String replyText, String exchange, String routingKey, AMQP.BasicProperties properties, byte[] body) throws IOException {
        System.out.println("Message returned with replyCode: " + replyCode);
        // 在这里处理消息发送到Queue失败的返回
    }
});
</code></pre>
  <p data-lake-id="u1b03a2e8" id="u1b03a2e8"><br></p>
  <p data-lake-id="u6c0ad104" id="u6c0ad104"><span data-lake-id="ub4cec00d" id="ub4cec00d">通过以上方式，我们注册了两个回调监听，用于在消息发送到Exchange或者Queue失败时进行异常处理。</span><strong><span data-lake-id="ufd5cea36" id="ufd5cea36">通常我们可以在失败时进行报警或者重试来保障一定能发送成功。</span></strong></p>
  <p data-lake-id="u3242abfd" id="u3242abfd"><br></p>
  <p data-lake-id="u3b855b3f" id="u3b855b3f"><span data-lake-id="u5e7009cd" id="u5e7009cd">完整的代码如下：</span></p>
  <p data-lake-id="u607eb267" id="u607eb267"><span data-lake-id="u4eb79677" id="u4eb79677">​</span><br></p>
  <pre lang="java"><code>
import com.rabbitmq.client.*;

public class PublisherCallbacksExample {

    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");

        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {

            // 启用Publisher Confirms
            channel.confirmSelect();

            // 设置Publisher Confirms回调
            channel.addConfirmListener(new ConfirmListener() {
                @Override
                public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                    System.out.println("Message confirmed with deliveryTag: " + deliveryTag);
                    // 在这里处理消息确认
                }

                @Override
                public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                    System.out.println("Message not confirmed with deliveryTag: " + deliveryTag);
                    // 在这里处理消息未确认
                }
            });

            // 启用Publisher Returns
            channel.addReturnListener(new ReturnListener() {
                @Override
                public void handleReturn(int replyCode, String replyText, String exchange, String routingKey, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    System.out.println("Message returned with replyCode: " + replyCode);
                    // 在这里处理消息发送到Queue失败的返回
                }
            });

            String exchangeName = "my_exchange";
            String routingKey = "my_routing_key";
            String message = "Hello, RabbitMQ!";

            // 发布消息到Exchange
            channel.basicPublish(exchangeName, routingKey, true, null, message.getBytes());

            // 等待Publisher Confirms
            if (!channel.waitForConfirms()) {
                System.out.println("Message was not confirmed!");
            }

            // 关闭通道和连接
            channel.close();
        }
    }
}

</code></pre>
  <p data-lake-id="uf078a075" id="uf078a075"><br></p>
  <p data-lake-id="u77343255" id="u77343255"><span data-lake-id="u26cbe15a" id="u26cbe15a">另外，这里如果发送到Queue之后，是否一定能持久化下来，是否一定不丢，这就是另外一个话题了。</span></p>
  <p data-lake-id="u58f73bdc" id="u58f73bdc"><span data-lake-id="u69724e28" id="u69724e28">​</span><br></p>
  <h1 data-lake-id="HvXFl" id="HvXFl"><br></h1>
 </body>
</html>